package com.hadwinling.alogriithm.projectforpso.miemiepso

/*PSO implements in Spark
 * population size:16
 * dimension:16
 * every partical:num,pos,vel,fitness,pb_pos,pb_fitness
 *
 */

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.util.Random

object SparkPSO {
  val pop_size = 16
  val iter_max = 10
  val dim = 16
  var t = 0 //loop counter
  /*initialize the global best :fitness,pos,vel*/
  var gb: (Double, (Array[Double], Array[Double])) = (Double.PositiveInfinity, (Array.fill(dim)(0.0), Array.fill(dim)(0.0)))

  def main(args: Array[String]): Unit = {
    val time_start = System.nanoTime
    val partitionsNum = 4
    //  val conf = new SparkConf().setAppName("SparkPSO").setMaster("spark://211.69.243.34:7077")
    val conf = new SparkConf().setAppName("SparkPSO").setMaster("local")

    val sc = new SparkContext(conf)
    /* initialize the pos and vel of every partical individual:(num,(pos,vel,pb_pos))*/
    val particals_tmp: RDD[(Int, (Array[Double], Array[Double], Array[Double]))] = sc.parallelize(List.range(0, pop_size), partitionsNum)
      .map(num => (num, (Array.fill(dim)(Random.nextDouble()), Array.fill(dim)(Random.nextDouble()), Array.fill(dim)(Random.nextDouble()))))
    var particles_fitness: RDD[(Int, (Array[Double], Array[Double], Double, Array[Double], Double))] = particals_tmp.mapPartitions(PSOStuff.addFitness).persist()
    //particlesBC: (num,(pos,vel,gb_pos))
    var particlesBC: Broadcast[collection.Map[Int, (Array[Double], Array[Double], Array[Double])]] = sc.broadcast(particles_fitness.map(x => (x._1, (x._2._1, x._2._2, x._2._4))).collectAsMap)


    /**
     * sortBykey 这一步是比较消耗性能的，可以考虑topk算法
     */
    gb = particles_fitness.map(x => (x._2._3, (x._2._1, x._2._2))).sortByKey(true).take(1)(0)
    //gbBC:(fitness,(pos,vel))
    /*find the global best */
    var gbBC: Broadcast[(Double, (Array[Double], Array[Double]))] = sc.broadcast(gb)
    t = t + 1
    /* calling the function and update the information of particles */
    var particlesArray: Array[(Int, (Array[Double], Array[Double], Array[Double]))] = sc.parallelize(List.range(0, pop_size), partitionsNum).map { x =>
      //      PSOStuff.updateParticle(x, particlesBC.value(1)._1, particlesBC.value(1)._2, particlesBC.value(1)._3, gbBC.value._2._1)
      //      更新粒子
      PSOStuff.updateParticle(x, particlesBC.value(x)._1, particlesBC.value(x)._2, particlesBC.value(x)._3, gbBC.value._2._1)
    }.collect
    //cleaning
    particlesBC.destroy

    while (t < iter_max) {
      particles_fitness = sc.parallelize(particlesArray, partitionsNum).mapPartitions(PSOStuff.addFitness).persist()
      particlesBC = sc.broadcast(particles_fitness.map(x => (x._1, (x._2._1, x._2._2, x._2._4))).collectAsMap)
      /*find the global best */
      gb = particles_fitness.map(x => (x._2._3, (x._2._1, x._2._2))).sortByKey(true).take(1)(0)
      gbBC.destroy
      gbBC = sc.broadcast(gb)
      //print the information of every particle in swarm of every interation
      List.range(0, pop_size).foreach { j =>
        println("num:" + particlesArray(j)._1)
        print("gb_pos:")
        List.range(0, dim).foreach { i =>
          print(gb._2._1(i) + "\t")
        }
        println
        println("gb_fitness:" + gb._1)
        print("pos:")
        List.range(0, dim).foreach { i =>
          print(particlesArray(j)._2._1(i) + "\t")
        }
        println
        print("vel:")
        List.range(0, dim).foreach { i =>
          print(particlesArray(j)._2._2(i) + "\t")
        }
        println
        print("pb_pos:")
        List.range(0, dim).foreach { i =>
          print(particlesArray(j)._2._3(i) + "\t")
        }
        println
        println
      }
      t = t + 1
      /*calling the function and update the information of particles*/
      particlesArray = sc.parallelize(List.range(0, pop_size), partitionsNum).map { x =>
        //          PSOStuff.updateParticle(x, particlesBC.value(1)._1, particlesBC.value(1)._2, particlesBC.value(1)._3, gbBC.value._2._1)
        PSOStuff.updateParticle(x, particlesBC.value(x)._1, particlesBC.value(x)._2, particlesBC.value(x)._3, gbBC.value._2._1)
      }.collect



      //cleaning
      particlesBC.destroy
    }
    val runtime = (System.nanoTime - time_start) / 1e9d
    //    println("Timer", runtime)
    //    println(sc.defaultParallelism)
    //    println(Random.nextDouble)
    sc.stop()
    //    particlesBC.value.foreach(println)
    //      println(gbBC.value._2._1)
    //    println(gbBC.id)
  }

}

